package com.example.minisqlite;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

public class Catalog implements Serializable {

    private int nextTableId = 1;
    private int nextIndexId = 1;
    private final Map<String, TableDefinition> tablesByName = new LinkedHashMap<>();
    private final Map<Integer, TableDefinition> tablesById = new LinkedHashMap<>();
    private final Map<String, IndexDefinition> indexesByName = new LinkedHashMap<>();
    private final Map<Integer, List<IndexDefinition>> indexesByTableId = new LinkedHashMap<>();

    public Collection<TableDefinition> getTables() {
        return Collections.unmodifiableCollection(tablesByName.values());
    }

    public TableDefinition getTable(String name) throws SQLException {
        TableDefinition def = tablesByName.get(name.toLowerCase());
        if (def == null) {
            throw new SQLException("Table not found: " + name);
        }
        return def;
    }

    public TableDefinition getTable(int tableId) throws SQLException {
        TableDefinition def = tablesById.get(tableId);
        if (def == null) {
            throw new SQLException("Unknown table id: " + tableId);
        }
        return def;
    }

    public TableDefinition createTable(String name, java.util.List<ColumnDefinition> columns) throws SQLException {
        String key = name.toLowerCase();
        if (tablesByName.containsKey(key)) {
            throw new SQLException("Table already exists: " + name);
        }
        int pkIndex = -1;
        for (int i = 0; i < columns.size(); i++) {
            if (columns.get(i).isPrimaryKey()) {
                if (pkIndex >= 0) {
                    throw new SQLException("Only one PRIMARY KEY is allowed");
                }
                pkIndex = i;
            }
        }
        TableDefinition def = new TableDefinition(nextTableId++, name, columns, pkIndex);
        tablesByName.put(key, def);
        tablesById.put(def.getTableId(), def);
        return def;
    }

    public IndexDefinition createIndex(String indexName, String tableName, String columnName) throws SQLException {
        String key = indexName.toLowerCase();
        if (indexesByName.containsKey(key)) {
            throw new SQLException("Index already exists: " + indexName);
        }
        TableDefinition table = getTable(tableName);
        int columnIndex = table.findColumnIndex(columnName);
        if (columnIndex < 0) {
            throw new SQLException("Unknown column for index: " + columnName);
        }
        IndexDefinition def = new IndexDefinition(nextIndexId++, indexName, table.getTableId(), columnName, columnIndex);
        indexesByName.put(key, def);
        indexesByTableId.computeIfAbsent(table.getTableId(), id -> new ArrayList<>()).add(def);
        return def;
    }

    public List<IndexDefinition> getIndexesForTable(int tableId) {
        return indexesByTableId.getOrDefault(tableId, List.of());
    }

    public IndexDefinition getIndex(String name) throws SQLException {
        IndexDefinition def = indexesByName.get(name.toLowerCase());
        if (def == null) {
            throw new SQLException("Index not found: " + name);
        }
        return def;
    }

    public byte[] serialize() throws IOException {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             ObjectOutputStream oos = new ObjectOutputStream(baos)) {
            oos.writeObject(this);
            oos.flush();
            return baos.toByteArray();
        }
    }

    public static Catalog deserialize(byte[] data) throws IOException, ClassNotFoundException {
        try (ByteArrayInputStream bais = new ByteArrayInputStream(data);
             ObjectInputStream ois = new ObjectInputStream(bais)) {
            return (Catalog) ois.readObject();
        }
    }
}
